#include "agg_operator.h"
#include <algorithm>

RC AggOperator::open()
{
  Operator * child = children_[0];
  child->open();

  ProjectTuple * projectTuple = static_cast<ProjectTuple *>(child->current_tuple());
  tuple_.speces_ = projectTuple->speces_;
  while(child->next() != RECORD_EOF) {
    Tuple * tuple = child->current_tuple();
    auto agg_Tuple = static_cast<ProjectTuple *>(tuple);
    all_tuples_.push_back(new ProjectTuple(*agg_Tuple));
  }
  // 对每个group by成员按升序排列
  std::sort(all_tuples_.begin(), all_tuples_.end(), [&](Tuple *t1, Tuple *t2){
    for(size_t i = 0; i < group_num_; i++){
      GroupAttr &group_attr = group_attributes_[i];
      TupleCell cell1, cell2;
      t1->find_cell(group_attr.relation_name, group_attr.attribute_name, cell1);
      t2->find_cell(group_attr.relation_name, group_attr.attribute_name, cell2);
      int res = cell1.compare(cell2);
      if(res != 0){
        return res < 0;
      }
    }
    return false;
  });
  return SUCCESS;
}

RC AggOperator::next()
{
  if(current_tuple_count_ >= (size_t)all_tuples_.size()){
    return RC::RECORD_EOF;
  }
  std::vector<Tuple *> tmp_tuples_;
  tmp_tuples_.push_back(all_tuples_[current_tuple_count_++]);
  while(current_tuple_count_ < (int32_t)all_tuples_.size() &&
      in_a_group(all_tuples_[current_tuple_count_ - 1], all_tuples_[current_tuple_count_])){
        tmp_tuples_.push_back(all_tuples_[current_tuple_count_++]);
  }
  tuple_.tuple_cells_.clear();
  // 一组一组聚合
  for (int i = 0; i < (int32_t)tuple_.speces_.size(); i++) {
    tuple_.set_tuple_cells(i, tmp_tuples_, tuple_.speces_[i]->alias());
  }

  return RC::SUCCESS;
}

bool AggOperator::in_a_group(Tuple *t1, Tuple *t2)
{
  for(size_t i = 0; i < group_num_; i++){
    GroupAttr &group_attr = group_attributes_[i];
    TupleCell cell1, cell2;
    t1->find_cell(group_attr.relation_name, group_attr.attribute_name, cell1);
    t2->find_cell(group_attr.relation_name, group_attr.attribute_name, cell2);
    int res = cell1.compare(cell2);
    if(res != 0){
      return false;
    }
  }
  return true;
}

Tuple *AggOperator::current_tuple()
{
  return &tuple_;
}

RC AggOperator::close()
{
  return children_[0]->close();
}
